Dataformのクライアントライブラリを使ってワークスペース内のファイルを全て取得してみた

pip install google-cloud-dataform

file: ".gitignore"
file: "dataform.json"
directory: "definitions"

import os
from google.cloud import dataform_v1beta1
from google.cloud.dataform_v1beta1.types import QueryDirectoryContentsRequest, ReadFileRequest
from google.api_core.exceptions import NotFound

def query_directory_recursive(client, workspace_id, path="", local_base_path="."):
    # リクエストパラメータを初期化
    request = QueryDirectoryContentsRequest(
        workspace=workspace_id,
        path=path,
    )
    try:
        # リクエストを実行して、ディレクトリの内容を取得
        page_result = client.query_directory_contents(request=request)
        for response in page_result:
            response_str = str(response)  # レスポンスを文字列として処理
            if response_str.startswith("file: "):
                # ファイルの場合
                file_path = response_str[len("file: "):].strip('" \n')
                print(f"Processing file: {file_path}")
                # ファイルの内容を読み取る
                read_file(client, workspace_id, file_path, local_base_path)
            elif response_str.startswith("directory: "):
                # ディレクトリの場合
                dir_path = response_str[len("directory: "):].strip('" \n')
                print(f"Processing directory: {dir_path}")
                # ローカルにディレクトリを作成
                os.makedirs(os.path.join(local_base_path, dir_path), exist_ok=True)
                # ディレクトリ内を再帰的に検索
                query_directory_recursive(client, workspace_id, dir_path, local_base_path)
    except NotFound as e:
        # ディレクトリが見つからない場合のエラーハンドリング
        print(f"Error: {e.message}")

def read_file(client, workspace_id, file_path, local_base_path):
    # ファイル読み取りリクエストを初期化
    request = ReadFileRequest(
        workspace=workspace_id,
        path=file_path,
    )
    # ファイルの内容を取得
    response = client.read_file(request=request)
    # ローカルのファイルパスを構築
    local_file_path = os.path.join(local_base_path, file_path)
    # 必要なディレクトリが存在することを確認
    local_dir = os.path.dirname(local_file_path)
    os.makedirs(local_dir, exist_ok=True)
    # ファイルの内容をローカルファイルに書き込む
    with open(local_file_path, 'wb') as f:
        f.write(response.file_contents)

if __name__ == "__main__":
    # 自身の環境を設定
    project_id = "プロジェクトID"
    location = "asia-northeast1"
    repository_id = "リポジトリ名"
    workspace_id = "ワークスペース名"
    # ワークスペースの文字列を構築
    workspace_str = f"projects/{project_id}/locations/{location}/repositories/{repository_id}/workspaces/{workspace_id}"
    local_base_path = "./local_copy"
    # Dataformクライアントを作成
    client = dataform_v1beta1.DataformClient()
    # ディレクトリの再帰的な検索を開始
    query_directory_recursive(client, workspace_str, local_base_path=local_base_path)

$ python get_workspace_contents.py 
Processing file: .gitignore
Processing file: dataform.json
Processing directory: definitions
Processing file: definitions/test_1.sqlx
Processing file: definitions/test_2.sqlx
Processing file: definitions/test_3.sqlx
...(以下略)

url = f"https://dataform.googleapis.com/v1beta1/projects/{project_id}/locations/{location}/repositories/{repository}:queryDirectoryContents"
headers = {
    "Authorization": f"Bearer {fetch_token()}"
}
params = {
    "path": directory
}

request = QueryDirectoryContentsRequest(
    workspace=workspace_id,
    path=path,
)
client.query_directory_contents(request=request)